package org.onion.pattern.notify.dingtalk;

import com.dingtalk.api.DefaultDingTalkClient;
import com.dingtalk.api.request.OapiRobotSendRequest;
import com.dingtalk.api.response.OapiRobotSendResponse;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import lombok.Data;
import lombok.Setter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.Base64;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
public class DingTalkRobotNotifyExecutor {

    private final LinkedBlockingQueue<Msg> queue;
    private final ExecutorService executorService;
    private final AtomicBoolean isClose;
    private final AtomicBoolean started;
    private final int queueSize;
    private final int threadSize;
    private final Cache<String, DefaultDingTalkClient> dingTalkClientCache;
    @Setter
    private String notifyUrl = "https://oapi.dingtalk.com/robot/send";

    /**
     * 默认的执行器
     */
    public final static DingTalkRobotNotifyExecutor EXECUTOR = new DingTalkRobotNotifyExecutor(1, 1000) {
        @Override
        public void setNotifyUrl(String notifyUrl) {
            throw new UnsupportedOperationException();
        }
    };


    DingTalkRobotNotifyExecutor(int threadSize, int queueSize) {
        this.queueSize = queueSize;
        this.threadSize = threadSize;
        this.queue = new LinkedBlockingQueue<>(queueSize);
        this.executorService = Executors.newSingleThreadExecutor();
        this.isClose = new AtomicBoolean(false);
        this.started = new AtomicBoolean(false);
        this.dingTalkClientCache = CacheBuilder.newBuilder()
                //配置最大容量为100，基于容量进行回收
                .maximumSize(200)
                //配置写入后多久使缓存过期
                .expireAfterWrite(3, TimeUnit.HOURS)
                //多久没有访问之后过期
//                .expireAfterAccess(10, TimeUnit.MINUTES)
                //配置写入后多久刷新缓存
//                .refreshAfterWrite(3, TimeUnit.SECONDS)
                //key使用弱引用-WeakReference
                .weakKeys()
                //当Entry被移除时的监听器-下文会讲述
//                .removalListener(notification -> System.out.println("notification=" + notification))
                //创建一个CacheLoader，重写load方法，以实现"当get时缓存不存在，则load，放到缓存并返回的效果
                .build();

    }


    @SneakyThrows
    private static String signURL(String url, String accessToken, String secret) {
        String timestamp = String.valueOf(Instant.now().toEpochMilli());
        String stringToSign = timestamp + "\n" + secret;
        Mac sha256_HMAC = Mac.getInstance("HmacSHA256");
        SecretKeySpec secret_key = new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), "HmacSHA256");
        sha256_HMAC.init(secret_key);
        byte[] signBytes = sha256_HMAC.doFinal(stringToSign.getBytes(StandardCharsets.UTF_8));
        String sign = Base64.getEncoder().encodeToString(signBytes);
        // 构造请求URL
        return String.format("%s?access_token=%s&timestamp=%s&sign=%s", url, accessToken, timestamp, URLEncoder.encode(sign, StandardCharsets.UTF_8.name()));
    }

    private void sendMsg() {
        while (!isClose.get() && !Thread.interrupted()) {
            try {
                Msg msg = queue.poll(3, TimeUnit.SECONDS);
                if (msg == null) {
                    continue;
                }
                log.trace("钉钉消息发送 {}", msg);
                //todo 消息过期处理 ttl

                DefaultDingTalkClient dingTalkClient = this.dingTalkClientCache.get(msg.getAccessToken(), ()
                        -> new DefaultDingTalkClient(String.format("%s?access_token=%s", notifyUrl, msg.getAccessToken())));

                //加签处理
                if (StringUtils.isNotBlank(msg.signKey)) {
                    dingTalkClient.resetServerUrl(signURL(notifyUrl, msg.getAccessToken(), msg.signKey));
                }

                OapiRobotSendResponse response = dingTalkClient.execute(msg.getRequest());

                if (!StringUtils.equalsIgnoreCase(response.getErrorCode(), "0")) {
                    if (!StringUtils.equalsIgnoreCase(response.getMsg(), "ok")) {
                        log.warn("【钉钉消息推送】 消息发送失败  body {}", response.getBody());
                    }
                }
            } catch (Exception e) {
                log.warn("【钉钉消息推送】 消息发送失败", e);
            }
        }
    }


    public void close() {
        this.isClose.set(true);
        try {
            this.executorService.shutdown();
        } catch (Exception e) {
            //
        }
        try {
            int size = this.queue.size();
            if (size > 0) {
                log.error("【钉钉消息推送】 容器将要关闭,还有 {} 条消息未推送", size);
            }
            this.queue.clear();
        } catch (Exception e) {
            //
        }
        this.dingTalkClientCache.cleanUp();
    }

    private synchronized void submitThread() {
        if (this.started.get()) {
            return;
        }

        //启动线程池
        this.executorService.submit(this::sendMsg);

        this.started.set(true);
    }

    public void write(Msg msg) {

        if (!this.started.get()) {
            submitThread();
        }

        try {

            if (queue.size() > this.queueSize) {
                log.warn("【钉钉消息推送】 消息溢出已丢弃,队列大小 {}", this.queueSize);
                return;
            }

            queue.put(msg);
        } catch (InterruptedException e) {
            log.warn("【钉钉消息推送】", e);
        }

    }

    @Data
    public static class Msg {
        private final long timestamp;
        private String accessToken;
        private String signKey;
        private OapiRobotSendRequest request;

        public Msg() {
            this.timestamp = System.currentTimeMillis();
        }
    }


}
